package flink.api_study.checkpoint;

import flink.api_study.source.Person;
import flink.api_study.source.PersonSource;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Map;
import java.util.Random;

public class FaultTolerance {

    public static void main(String[] args) throws Exception{

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 传参
        ParameterTool parameters = ParameterTool.fromArgs(args);
        //将ParameterTool的访问范围设置为global
        env.getConfig().setGlobalJobParameters(parameters);
        env.setParallelism(1);

        // 1. 状态后端配置
//        env.setStateBackend( new MemoryStateBackend());
        env.setStateBackend( new FsStateBackend( "file:///home/flink/checkpoint"));//linux
//        env.setStateBackend( new FsStateBackend( "file:///"+System.getProperty("user.dir")+ "\\src\\main\\resources\\ck\\"));//windows
//        env.setStateBackend( new RocksDBStateBackend(""));

        // 2. 检查点配置
        env.enableCheckpointing(300);
        //模式选择设置
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //设置超时时间,超时时间是说，这个检查点设置了，但是过去60000ms还没有完成就认为这个检查点超时了，然后把这个检查点终止，也就是不再使用这个检查点了
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        /**设置检查点尝试之间的最小间隔。此设置定义检查点协调器在可能触发另一个检查点（相对于最大并发检查点数）后多久可以触发另一个检查点
         * 比如下面的设置，下一个检查点将在上一个检查点完成后，不超过5s内启动
         * 这个参数跟检查点生成的间隔时间是有点冲突的，这个参数不考虑上面提到的检查点持续时间和检查点间隔。或者说设置下面的参数以后，检查点自动就变成5s了，
         * 如果觉得5s不够可以再将间隔设置的大一点，但是不能小于5s了
         * 设置检查点的interval和设置检查点之间的间隔时间有啥不同呢？
         * interval是物理时间的间隔，即时间只要过去1s了，就会生成一个检查点。但是设置检查点之间的间隔是说检查点完成1s了就会设置间隔，这个是跟检查点完成时间相关的
         * 比如存储检查点的系统比较慢，完成一个检查点平均10s，然后下面检查点之间的间隔设置为5s，那么两个检查点生成的时间间隔就是15s
         * 再简单点说，interval设置的是两个检查点生成时刻的间隔，而下面参数设置的是第一个检查点结束和第二个检查点创建（还没有结束）之间的间隔
         **/
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
        //设置检查点最大并发数.比如设置为1，表示只能有一个检查点，当这个检查点完成以后才有可能产生下一个检查点。这也是默认参数。
        // 如果定义了上面的设置，则不能再定义这条设置项大于1，否则运行时报错
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        //检查点和保存点都可以用于容灾，这个配置是说如果有一个时间更近的保存点，我也不用，而更加愿意使用检查点进行容灾
        env.getCheckpointConfig().setPreferCheckpointForRecovery(true);
        //设置使用非对齐检查点。它可以极大减少背压情况下检查点的时间，但是只有在精确一次的检查点并且允许的最大检查点并发数量为1的情况下才能使用
        env.getCheckpointConfig().enableUnalignedCheckpoints();
        //检查点失败容忍值
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(0);
        //RETAIN_ON_CANCELLATION：在任务取消时保留检查点，此时就需要手动清理无用的检查点。
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        // 3. 重启策略配置
        // 固定延迟重启
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000L));
        // 失败率重启(最大失败次数,衡量失败次数的是时间段,间隔)
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.minutes(10), Time.minutes(1)));
        //无重启配置
        env.setRestartStrategy(RestartStrategies.noRestart());


        DataStream<Person> dataStream = env.addSource(new PersonSource());
        dataStream.keyBy(person -> person.getJob())
                .map(new MyMapFuction1())
                .print();
        dataStream.print();
        env.execute();
    }

    static class MyMapFuction1 extends RichMapFunction<Person, Integer> {
        private ValueState<Integer> keyCountState;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            keyCountState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("key-count",Integer.class,0));
        }

        @Override
        public Integer map(Person person) throws Exception {
            Integer count = keyCountState.value();

            //失败测试
            Map<String,String> globalParams = getRuntimeContext().getExecutionConfig().getGlobalJobParameters().toMap();
            String fail = globalParams.get("fail");
            if(StringUtils.isNotBlank(fail) && fail.equals("1")){
                if(new Random().nextInt(10) == 1){//四分之一概率失败
                    throw new RuntimeException("主动失败");
                }
            }

            count++;
            keyCountState.update(count);
            return count;
        }
    }
}
